package com.qupeng.demo.kafka.kafkaspringbootproducer.avro;

import org.apache.avro.Schema;
import org.apache.avro.io.*;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.avro.specific.SpecificDatumWriter;

import java.io.ByteArrayOutputStream;

/**
 * avro工具类
 */
public class MyAvroUtil {

    /**
     * 根据不同类型的数据AVRO序列化为字节数组
     * @param data 泛型数据
     * @param schema 模式
     * @param <T> 泛型
     * @return byte[] 字节数组
     */
    public static <T> byte[] serializeData(T data, Schema schema) {
        try (ByteArrayOutputStream outputStream = new ByteArrayOutputStream()){
            ;
            DatumWriter<T> writer = new SpecificDatumWriter<>(schema);
            BinaryEncoder binaryEncoder = EncoderFactory.get().directBinaryEncoder(outputStream, null);
            writer.setSchema(schema);
            writer.write(data, binaryEncoder);
            outputStream.flush();
            //序列化后的字节数组
            return outputStream.toByteArray();
        }catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 根据字节数组反序列化为指定类型
     *
     * @param cls 指定类型
     * @param schema 模式
     * @param byteArray 字节数组
     * @param <T> 泛型
     * @return T 具体类型数据
     */
    public static <T>T deSerializeData(Class<T> cls, Schema schema, byte[] byteArray) {
        try {
            T data = null;
            DatumReader<T> datumReader = new SpecificDatumReader<T>(schema);
            BinaryDecoder binaryDecoder = DecoderFactory.get().binaryDecoder(byteArray, null);
            datumReader.setSchema(schema);
            if (!binaryDecoder.isEnd()) {
                data = datumReader.read(null, binaryDecoder);
            }
            return data;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

